package simpledb;

import java.util.*;

/**
 * Knows how to compute some aggregate over a set of IntFields.
 */
public class IntegerAggregator implements Aggregator {

    private static final long serialVersionUID = 1L;

    private int groupByField;
    private Type groupByFieldType;
    private int aggregateField;
    private Aggregator.Op op;

    private TupleDesc tupleDesc;
    private HashMap hashMap = new HashMap();
    private HashMap avgCount;
    private Set set;
    private Iterator iterator;

    /**
     * Aggregate constructor
     *
     * @param gbfield     the 0-based index of the group-by field in the tuple, or
     *                    NO_GROUPING if there is no grouping
     * @param gbfieldtype the type of the group by field (e.g., Type.INT_TYPE), or null
     *                    if there is no grouping
     * @param afield      the 0-based index of the aggregate field in the tuple
     * @param what        the aggregation operator
     */

    public IntegerAggregator(int gbfield, Type gbfieldtype, int afield, Op what) {
        // some code goes here
        this.groupByField = gbfield;
        this.groupByFieldType = gbfieldtype;
        this.aggregateField = afield;
        this.op = what;
        if (what.equals(Op.AVG)){
            avgCount=new HashMap();
        }
    }

    /**
     * Merge a new tuple into the aggregate, grouping as indicated in the
     * constructor
     *
     * @param tup the Tuple containing an aggregate field and a group-by field
     */
    @Override
    public void mergeTupleIntoGroup(Tuple tup) {
        // some code goes here
        if (groupByField==Aggregator.NO_GROUPING){
            mergeTupleWithoutGroup(tup);
        }else {
            mergeTupleWithGroup(tup);
        }

    }

    private void mergeTupleWithGroup(Tuple tup){
        Field key = tup.getField(groupByField);
        IntField value1 = (IntField) tup.getField(aggregateField);
        if (hashMap.isEmpty()) {
            Type[] types = {tup.getTupleDesc().getFieldType(groupByField),
                    tup.getTupleDesc().getFieldType(aggregateField)};
            tupleDesc = new TupleDesc(types);
        }
        if (!hashMap.containsKey(key)) {
            if (op.equals(Op.COUNT)) {
                hashMap.put(key, 1);
            } else if (op.equals(Op.SC_AVG)) {

            } else if (op.equals(Op.SUM_COUNT)) {

            } else if (op.equals(Op.AVG)) {
                hashMap.put(key, value1);
                avgCount.put(key, 1);
            } else {
                hashMap.put(key, value1);
            }
            return;
        }

        if (op.equals(Op.MAX)) {
            IntField value2 = (IntField) hashMap.get(key);
            if (value1.compare(Predicate.Op.GREATER_THAN,value2)) {
                hashMap.remove(key);
                hashMap.put(key, value1);
            }
        } else if (op.equals(Op.MIN)) {
            IntField value2 = (IntField) hashMap.get(key);
            if (value1.compare(Predicate.Op.LESS_THAN,value2)) {
                hashMap.remove(key);
                hashMap.put(key, value1);
            }
        } else if (op.equals(Op.COUNT)) {
            int cnt = (int) hashMap.get(key);
            cnt++;
            hashMap.remove(key);
            hashMap.put(key, cnt);
        } else if (op.equals(Op.AVG)) {
            IntField value2 = (IntField) hashMap.get(key);
            int cnt=(int)avgCount.get(key);
            int out=value1.getValue()+value2.getValue();
            IntField value=new IntField(out);

            avgCount.remove(key);
            avgCount.put(key, cnt+1);
            hashMap.remove(key);
            hashMap.put(key, value);
        } else if (op.equals(Op.SUM)) {
            IntField value2 = (IntField) hashMap.get(key);
            int out=value1.getValue()+value2.getValue();
            Field value=new IntField(out);

            hashMap.remove(key);
            hashMap.put(key, value);
        } else if (op.equals(Op.SC_AVG)) {

        } else if (op.equals(Op.SUM_COUNT)) {

        }
        return;
    }

    private void mergeTupleWithoutGroup(Tuple tup){
        IntField value1 = (IntField) tup.getField(aggregateField);
        if (hashMap.isEmpty()){
            Type[] types={tup.getTupleDesc().getFieldType(aggregateField)};
            tupleDesc = new TupleDesc(types);
            switch (this.op){
                case AVG -> {
                    avgCount.put(1,1);
                    hashMap.put(Aggregator.NO_GROUPING,value1);
                    break;
                }case MAX -> {
                    hashMap.put(Aggregator.NO_GROUPING,value1);
                    break;
                }case MIN -> {
                    hashMap.put(Aggregator.NO_GROUPING,value1);
                    break;
                }case SUM -> {
                    hashMap.put(Aggregator.NO_GROUPING,value1);
                    break;
                }case COUNT -> {
                    hashMap.put(Aggregator.NO_GROUPING,1);
                    break;
                }case SC_AVG -> {
                    break;
                }case SUM_COUNT -> {
                    break;
                }default -> {
                    throw new IllegalArgumentException("Unknown Aggregate");
                }
            }
        }else {
            int key=Aggregator.NO_GROUPING;
            switch (this.op){
                case AVG -> {
                    int cnt= (int) avgCount.get(1);
                    cnt++;
                    IntField value2= (IntField) hashMap.get(Aggregator.NO_GROUPING);
                    IntField value=new IntField(value1.getValue()+value2.getValue());

                    avgCount.remove(1);
                    avgCount.put(1,cnt);
                    hashMap.remove(Aggregator.NO_GROUPING);
                    hashMap.put(Aggregator.NO_GROUPING,value);
                    break;
                }case MAX -> {
                    IntField value2 = (IntField) hashMap.get(key);
                    if (value1.compare(Predicate.Op.GREATER_THAN,value2)) {
                        hashMap.remove(key);
                        hashMap.put(key, value1);
                    }

                    break;
                }case MIN -> {
                    IntField value2 = (IntField) hashMap.get(key);
                    if (value1.compare(Predicate.Op.LESS_THAN,value2)) {
                        hashMap.remove(key);
                        hashMap.put(key, value1);
                    }
                    break;
                }case SUM -> {
                    IntField value2 = (IntField) hashMap.get(key);
                    int out=value1.getValue()+value2.getValue();
                    Field value=new IntField(out);

                    hashMap.remove(key);
                    hashMap.put(key, value);
                    break;
                }case COUNT -> {
                    int cnt = (int) hashMap.get(key);
                    cnt++;
                    hashMap.remove(key);
                    hashMap.put(key, cnt);
                    break;
                }case SC_AVG -> {
                    break;
                }case SUM_COUNT -> {
                    break;
                }default -> {
                    throw new IllegalArgumentException("Unknown Aggregate");
                }
            }
        }
    }

    /**
     * Create a OpIterator over group aggregate results.
     *
     * @return a OpIterator whose tuples are the pair (groupVal, aggregateVal)
     * if using group, or a single (aggregateVal) if no grouping. The
     * aggregateVal is determined by the type of aggregate specified in
     * the constructor.
     */
    @Override
    public OpIterator iterator() {
        // some code goes here
        return new OpIterator() {
            @Override
            public void open() throws DbException, TransactionAbortedException {
                set=hashMap.entrySet();
                iterator=set.iterator();
            }

            @Override
            public boolean hasNext() throws DbException, TransactionAbortedException {
                // System.out.println("IntegerAggregator.iterator.hasNext");
                return iterator.hasNext();
            }

            @Override
            public Tuple next() throws DbException, TransactionAbortedException, NoSuchElementException {
                Map.Entry map= (Map.Entry) iterator.next();
                Tuple tuple=new Tuple(tupleDesc);
                if (groupByField==Aggregator.NO_GROUPING){
                    if (op.equals(Op.AVG)){
                        int cnt= (int) avgCount.get(1);
                        IntField intField= (IntField) map.getValue();
                        int out=(int) intField.getValue()/cnt;
                        tuple.setField(0,new IntField(out));
                    }else {
                        tuple.setField(0,(Field) map.getValue());
                    }
                }else {
                    tuple.setField(0, (Field) map.getKey());
                    if (op.equals(Op.COUNT)){
                        tuple.setField(1, new IntField((int)map.getValue()));
                    }else if (op.equals(Op.AVG)){
                        IntField tolField =(IntField)map.getValue();
                        int tol=tolField.getValue();
                        int cnt=(int)avgCount.get((Field)map.getKey());
                        tuple.setField(1,new IntField((int) tol/cnt));
                    } else {
                        tuple.setField(1, (Field) map.getValue());
                    }
                }
                return tuple;
            }

            @Override
            public void rewind() throws DbException, TransactionAbortedException {
                this.close();
                this.open();
            }

            @Override
            public TupleDesc getTupleDesc() {
                return tupleDesc;
            }

            @Override
            public void close() {
                iterator=null;
            }
        };
    }

}
